資訊檢索 (Information Retrieval, IR) 在討論如何快速的搜尋使用者想要找到的結果，在 LLM 出現之前已經是個相對成熟的領域，我們天天在用的 Google Search 就是最經典的一種資訊檢索應用。
想要建立一個能夠快速龐大文檔的檢索引擎，首先需要建立索引，方法包含關鍵字、向量空間或 BM25 等等。接著根據使用者的查詢，尋找相關資訊並進行排名，這整個搜尋的模式，被稱為檢索模型 (Retrieval Model)。
將 LLM 與資訊檢索結合，不僅能讓語言模型接觸到外部知識，使其能回答不僅限於訓練資料內的內容，也能進一步與上下文學習 (In-Context Learning) 結合，使模型能夠更精準的進行分類。
今天就來介紹 Retrieval 相關的基本知識吧！
(Powered By Microsoft Designer)
關鍵字索引是個簡單有效卻經常被忽略的做法，雖然中文的關鍵字會遇到斷詞與邊界的問題，但是在簡單實做上，我們可以透過 Character-Based 的方式來處理。其中最容易被用來實做關鍵字索引的方法就是 N-Gram 索引。
N-Gram 索引的概念很單純，我們透過一個字一個字抓一個索引、兩個字兩個字抓一個索引的方式來建立索引，例如：
文本 A：今天天氣真好
文本 B：今天出門可好
他們的索引可能就會建成：
今、天、天、氣、真、好、今天、天天、天氣、氣真、真好、今天天、...
今、天、出、門、可、好、今天、天出、出門、門可、可好、今天出、...
用這些索引與文本對應起來：
今: A, B
天: A, B
氣: A
門: B
今天: A, B
天天: A
天出: B
依此類推，雖然會切出很多不成詞彙的片段，但是在小文本的情況下影響通常不大。
我們可以透過以下程式來列舉 N-Gram 結果：
def ngram(text: str, n: int):
for i in range(0, len(text) - n + 1):
yield text[i : i + n]
def all_ngram(text: str, a, b, step):
a = max(1, a)
b = min(len(text), b)
for i in range(a, b + 1, step):
for seg in ngram(text, i):
yield seg
segs = [seg for seg in all_ngram("今天天氣真好", 1, 3, 1)]
print(segs)
# ['今', '天', '天', '氣', '真', '好', '今天', '天天', '天氣', '氣真', '真好', '今天天', '天天氣', '天氣真', '氣真好']
segs = [seg for seg in all_ngram("今天天氣真好", 2, 6, 2)]
print(segs)
# ['今天', '天天', '天氣', '氣真', '真好', '今天天氣', '天天氣真', '天氣真好', '今天天氣真好']
一個關鍵字可能會索引出很多片段，可以考慮使用 Jaccard Similarity 做排序，也就是將彼此 N-Gram 的交集除以聯集：
def calc_jaccard(query: str, chunk: str):
query_length = len(query)
query_ngrams = {s for s in all_ngram(query, 1, query_length)}
chunk_ngrams = {s for s in all_ngram(chunk, 1, query_length)}
inter = query_ngrams & chunk_ngrams
union = query_ngrams | chunk_ngrams
return len(inter) / len(union)
最後完整的 N-Gram Search 大致如下：
import json
import os
from collections import defaultdict
def main():
# 建立 N-Gram 索引
index = defaultdict(set)
chunks = list()
for full_path in iter_markdown("."):
with open(full_path, "rt", encoding="UTF-8") as fp:
# 以雙換行作為段落邊界
segments = fp.read().split("\n\n")
for chunk in segments:
# 將所有 N-Gram 都當成索引
# 並將 Chunk ID 放進索引對應的列表
for seg in all_ngram(chunk, 1, 6):
index[seg].add(len(chunks))
chunks.append(chunk)
# 將索引與區塊存下來
with open("index.json", "wt", encoding="UTF-8") as fp:
_index = {k: list(index[k]) for k in index}
json.dump(_index, fp, ensure_ascii=False)
with open("chunks.json", "wt", encoding="UTF-8") as fp:
json.dump(chunks, fp, ensure_ascii=False)
# 開始實際查詢
query = "語言模型"
q_len = len(query)
query_ngrams = {s for s in all_ngram(query, 1, len(query))}
results = []
for i in index[query]:
# 計算 Jaccard Similarity 當作排名依據
score = calc_jaccard(query_ngrams, q_len, chunks[i])
results.append((score, chunks[i]))
results = sorted(results, reverse=True)
# 輸出前五名的結果
for i, (score, res) in zip(range(5), results):
print(f"Rank {i}, Score: {score:.4f}, Chunk: {repr(res)}")
def iter_markdown(target_dir):
# 拜訪所有 Markdown 文件
for dir_path, _, file_list in os.walk(target_dir):
for file_name in file_list:
if not file_name.endswith(".md"):
continue
yield os.path.join(dir_path, file_name)
def ngram(text: str, n: int):
# 取得指定大小的 N-Gram
for i in range(0, len(text) - n + 1):
yield text[i : i + n]
def all_ngram(text: str, a, b, step=1):
# 取得範圍內所有大小的 N-Gram
a = max(1, a)
b = min(len(text), b)
for i in range(a, b + 1, step):
for seg in ngram(text, i):
yield seg
def calc_jaccard(query_ngrams, q_len, chunk: str):
# 計算 Jaccard Similarity
chunk_ngrams = {seg for seg in all_ngram(chunk, 1, q_len)}
inter = query_ngrams & chunk_ngrams
union = query_ngrams | chunk_ngrams
return len(inter) / len(union)
if __name__ == "__main__":
main()
在應付少量文本時，使用 N-Gram 建立索引通常不會有太大的問題。但是當文本量逐漸上升時，索引本身所佔用的記憶體就會是個大問題了。加上這種索引只能做文字比對，並沒有辦法進行語意比較，因此並不是最理想的做法。
BM25 (Best Matching) 是一種基於 TF-IDF 的統計排名方法，雖然是個古老的演算法，但是相當有效。我們可以透過 Rank-BM25 套件輕鬆使用這個演算法，參考程式碼：
from rank_bm25 import BM25Okapi
corpus = [
"這是第一篇文檔",
"這是第二篇文檔",
"文檔的內容很重要",
]
# 簡單使用 Character-Based 斷詞
corpus_chars = [[ch for ch in doc] for doc in corpus]
bm25 = BM25Okapi(corpus_chars)
query = "重要"
query_chars = [ch for ch in query]
# 計算 BM25 分數
scores = bm25.get_scores(query_chars)
print(f"Scores: {scores}")
# 輸出查詢結果
best = scores.argmax()
print(f"Query: {query}")
print(f"Result: {corpus[best]}")
BM25 是一個純粹的排名演算法，因此可以與其他 Retrieval 方法做結合，例如先用 N-Gram 或 Embedding 進行模糊搜尋，再用 BM25 做排序之類的。
先前已經介紹過 Embedding 是一種文字向量化的結果，其中最常見的有 Word Embedding 與 Sentence Embedding，通常 Sentence Embedding 也是透過 Word Embedding 算來的，即便是 Transformer LM，也是將 Sentence 切成若干 Tokens 之後取其 Word Embedding 後，再透過數個 Transformer Layers 計算出來的 Sentence Embedding。
一般來說，在 IR 裡面使用的都是 Sentence Embedding，可以將不同長度的句子都 Encode 成固定維度的向量，這樣就可以簡單的透過一個矩陣運算來解決問題。
取得 Sentence Embedding 的方式有很多種，但是有支援中文或多語的選擇還真的挺少。其中一個選擇是 Google 的 Universal Sentence Encoder 系列，其中 USE-Multilingual Large 是筆者較推薦的版本，使用方法如下：
# pip install tensorflow tensorflow_hub tensorflow_text
import numpy as np
import tensorflow as tf
import tensorflow_hub as hub
import tensorflow_text
# 不讓 Tensorflow 使用 GPU
tf.config.set_visible_devices([], "GPU")
# 下載 & 讀取模型
hub_url = "https://tfhub.dev/google/universal-sentence-encoder-multilingual-large/3"
# 也可以把模型下載下來後，直接指定 Local Path 讀取
model = hub.load(hub_url)
# 取得 Embedding
text = ["cat", "貓", "고양이", "dog", "狗", "개"]
embeddings = model(text)
# 計算內積作為相似度
similarity = np.inner(embeddings, embeddings)
np.set_printoptions(precision=2)
print(similarity)
可以看到 "cat", "貓", "고양이" 之間的內積較大，代表他們在語意上較為相近。
比較容易遇到的問題是，如果在 Tensorflow 與 PyTorch 混用的專案裡面，先讀取 Tensorflow 可能會造成 GPU 衝突的問題，解決方法之一是先匯入 PyTorch 相關的套件，或者透過 tf.config.set_visible_devices([], "GPU") 來停用 USE 使用 GPU 進行運算，在沒有 GPU 的情況下 USE 也是可以算的很快。
另外，雖然 tensorflow_text 看似沒有用到，但是如果不 import 他就會報錯，所以記得要 import tensorflow_text 才能用。
還有一個選擇是使用 Sentence Transformers 模型，在 Hugging Face Hub 上可以找到很多相關的模型，其用法如下：
# pip install sentence_transformers
import numpy as np
from sentence_transformers import SentenceTransformer
# 下載 & 讀取模型
model = SentenceTransformer("intfloat/multilingual-e5-base")
# 取得 Embedding
text = ["cat", "貓", "고양이", "dog", "狗", "개"]
embeddings = model.encode(text, normalize_embeddings=True)
# 計算內積作為相似度
similarity = np.inner(embeddings, embeddings)
np.set_printoptions(precision=2)
print(similarity)
這些 Encoder 的用法基本上大同小異，建議根據自身應用評估哪個模型的效果較合適。
衡量向量相似度的方法除了內積以外，還有像是 Cosine Similarity 以及 Euclidean Distance 等等。
歐基里得距離 (Euclidean Distance) 其實就是俗稱的「相減平方開根號」，也就是求兩點座標之間的直線距離，對應到高維向量來說，就是對每一個維度都計算相減平方開根號，然後再加總起來。
from sklearn.metrics.pairwise import euclidean_distances
dist = euclidean_distances(embeddings, embeddings)
餘弦相似度 (Cosine Similarity) 是在計算兩個向量之間的夾角，並對這個角度取餘弦。當兩個向量很接近的時候，他們的夾角就會很小，那他們的 Cosine 值就會越大。
cos(0°) = 1, cos(90°) = 0, cos(180°) = -1
在 Scikit-Learn 裡面也有 Cosine Distance 的函數，基本上就是 1 - Cosine Similarity，兩者的程式碼如下：
from sklearn.metrics.pairwise import (
cosine_distances,
cosine_similarity
)
dist = cosine_distances(embeddings, embeddings)
sim = cosine_similarity(embeddings, embeddings)
最後別忘了，相似度 (Similarity) 是越大代表語意越相近，而距離 (Distance) 則是越小代表越相近。
Facebook AI Similarity Search (Faiss) 是 Facebook 開發的一套向量搜尋套件，可以在非常高維度的向量裡面進行非常快速的搜索，是個相當有效率的工具。首先透過 pip 安裝套件：
pip install faiss-cpu
若要使用 GPU 版則需要透過 Conda 安裝：
conda install -c conda-forge faiss-gpu
Faiss 的用法相當簡潔：
import faiss
import numpy as np
# 定義向量集合
value = [
[1, 0, 0],
[0, 1, 0],
[0, 0, 1],
[1, 0, 1],
[0, 1, 1],
]
value = np.array(value)
# 建立向量索引
index = faiss.IndexFlatL2(3)
index.add(value)
# 開始查詢
query = np.array([[1, 0, 0]])
dist, indices = index.search(query, 3)
# 輸出距離與最近的向量
print(dist)
for i in indices[0]:
print(value[i])
可以透過 faiss.write_index 以及 faiss.read_index 來寫入與讀取索引向量：
faiss.write_index(index, "index.faiss")
index = faiss.read_index("index.faiss")
print(type(index)) # <class 'faiss.swigfaiss_avx2.IndexFlat'>
Faiss 除了進行一般的基本向量搜尋以外，還有很多其他優化的方法，專門處理超級大量且高維的向量，可以參考官方文件。在一般的小型應用裡面，通常以上用法就已經足夠了。
今天介紹了一些資訊檢索領域相關的小知識，其實這些技術的發展已經相當成熟，只是一直沒有一個很好的輸出窗口。如今有了 LLM 這個強大的 NLG 系統後，與 IR 的搭配使其更為錦上添花，也是現在許多 LLM 相關應用的關鍵技術。
接下來會介紹 LLM 在任務導向對話上的應用，明天見！
iThome鐵人賽
看影片追技術